package com.eversec.msg.config;

import com.eversec.kafka.consumer.KafkaConsumerServer;
import com.eversec.kafka.exception.KafkaInitException;
import com.eversec.kafka.producer.KafkaProducerServer;
import com.eversec.msg.util.props.PropertiesInfo;
import org.apache.log4j.Logger;

import java.util.List;

/**
 * Created by eversec on 2018/1/22.
 */
public class SendMsgToKF {
    private static Logger logger=Logger.getLogger(SendMsgToKF.class);

    private static KafkaProducerServer<List<String>> kafkaProducerServer=null;
    /**
     * 单例模式：如果每次发生数据都创建一个tcp 这样连接数会很多
     * @return
     * @throws KafkaInitException
     */
    public static KafkaProducerServer<List<String>> getKafkaProducerServer() throws KafkaInitException{
        if(kafkaProducerServer==null){
            kafkaProducerServer=new KafkaProducerServer<List<String>>(PropertiesInfo.KAFKA_PRODUCER_HOST);
        }
        return kafkaProducerServer;
    }

    /**
     * 发送数据到中央设备告警的主题
     * @param list
     */
    public  static void setToAlarmtype(List<String> list,String type) {
        try {
            getKafkaProducerServer().sendMessages(type, null, list);
        } catch (KafkaInitException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
//		try {
//			new KafkaProducerServer<Collection<String>>(GetPropertiesInfo.KAFKAIPANDPORT).sendMessages(type, null, list);
//		} catch (KafkaInitException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
    }

//    /**
//     * 引擎扫描结果反馈，引擎侧可以通过结果反馈接口（Report），将样本扫描结果发布到消息服务器（KAFKA）中的话题（Report Topic）,平台侧订阅话题，消费已发布的消息，对扫描结果进行清洗、重组
//     * @param consumerhost 消费者服务地址
//     * @param topic  会话
//     * @throws KafkaInitException
//     * @throws IOException
//     * @throws ClientProtocolException
//     */
//    public static void getMsAlarmType(INmAlarmBeanService instanceNmAlarmBeanService,String type) {
//        KafkaConsumerServer kafkaConsumerServer = null;
//        try {
//            kafkaConsumerServer = new KafkaConsumerServer(GetPropertiesInfo.KAFKA_CONSUMER_HOST,type);
//        } catch (KafkaInitException e) {
//            logger.error("创建kafka消费者失败,获取kafka消息异常返回", e);
//            return;
//        }
//        logger.info("创建消费者");
//        kafkaConsumerServer.start();
//        while(true){
//            List<String> messages=kafkaConsumerServer.getMessages();
//            //调用研判报告JSON解析方法
//            for (String message : messages) {
//                //需要判断预警级别，需要解析json数据
//                JSONObject jsStr = JSONObject.fromObject(message);
//                JSONObject alarmhb = new JSONObject();
//                //1紧急、2严重、3重要、4普通、只要有告警就会发生弹框
//                if(!jsStr.get("alarmLevel").equals("5")){
//                    alarmhb.put("equipmentName", jsStr.get("equipmentName"));
//                    alarmhb.put("alarmContent", jsStr.get("alarmContent"));
//                    try {
//                        HttpClientPostUtil.doPost(GetPropertiesInfo.DWPATH,alarmhb.toString());
//                    } catch (ClientProtocolException e) {
//                        logger.error("发送dwr消息至portal Http异常", e);
//                    } catch (IOException e) {
//                        logger.error("发送dwr消息至portal IO异常", e);
//                    }
//                }
//
//                //解析message 转成bean对象
//                JSONObject josnmessage = JSONObject.fromObject(message);
//                NmAlarmBean messagebean = (NmAlarmBean) JSONObject.toBean(josnmessage, NmAlarmBean.class);
//                try {
//                    //保存数据到数据表
////						SaveBean task = new SaveBean();
//                    SaveBean.saveBean(messagebean,instanceNmAlarmBeanService);
//                } catch (Exception e) {
//                    logger.error("保存告警信息失败," + e.getMessage());
//                }
//
//            }
//            try {
//                Thread.sleep(30000);
//            } catch (InterruptedException e) {
//                // TODO Auto-generated catch block
//                e.printStackTrace();
//            }
//        }
//    }



    /**
     * 消费者
     * @param topic
     */
    public static KafkaConsumerServer getKafkaConsumerServerByTopic(String topic) {
        KafkaConsumerServer kafkaConsumerServer = null;
        try {
            kafkaConsumerServer = new KafkaConsumerServer("",topic);
        } catch (KafkaInitException e) {
            logger.error("创建kafka消费者失败,获取kafka消息异常返回", e);
        }
        logger.info("创建消费者");
//		kafkaConsumerServer.start();
        return kafkaConsumerServer;
    }





    public static void main(String[] args) throws KafkaInitException {

        KafkaConsumerServer kafkaConsumerServer=new KafkaConsumerServer(PropertiesInfo.KAFKA_CONSUMER_HOST,"wjbTest");
        kafkaConsumerServer.start();
        while(true){
            //调用研判报告JSON解析方法
            List<String> messages=kafkaConsumerServer.getMessages();
            for (String message : messages) {
                //需要判断预警级别，需要解析json数据
//				JSONObject jsStr = JSONObject.fromObject(message);
//				JSONObject alarmhb = new JSONObject();
                System.out.println("======123========"+message);
                System.out.println("=======456======="+message);


            }
        }

    }





}

